//! @Author: DengLibin
//! @Date: Create in 2023-11-01 16:13:22
//! @Description:

use std::{
    fmt::Display,
    sync::atomic::{AtomicU32, Ordering},
};

use app_error::AppError;
use config::AppConfig;
use rust_common::{dir, file_util, log_util, mysql, sqlite};

use log::{error, info};
use tokio::runtime::Runtime;
pub mod app_error;
pub mod config;
//保存历史记录
static HISTORY_TXT: &str = "history.txt";

//转换结果
fn check_result<T, E: Display>(result: Result<T, E>) -> Result<T, AppError> {
    return match result {
        Ok(r) => Result::Ok(r),
        Err(err) => Result::Err(AppError {
            msg: err.to_string(),
        }),
    };
}

/// @Author: DengLibin
/// @Date: Create in 2024-01-08 17:42:49
/// @Description: 创建异步运行时
fn create_runtime() -> Result<Runtime, AppError> {
    // 创建单一线程的runtime
    // let rt_r: Result<tokio::runtime::Runtime, std::io::Error> = tokio::runtime::Builder::new_current_thread()
    //     .enable_io()
    //     .enable_time()
    //     .build();

    //多线程
    let rt_r = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(8) // 8个工作线程
        .global_queue_interval(61)
        .max_blocking_threads(48) //运行阻塞任务的最大线程数 不影响核心线程
        .thread_name("my-woker-thread")
        .enable_io() // 可在runtime中使用异步IO
        .enable_time() // 可在runtime中使用异步计时器(timer)
        .build(); // 创建runtime

    check_result(rt_r)
}

/// @Author: DengLibin
/// @Date: Create in 2023-11-01 17:07:47
/// @Description:
fn sqlite2mysql(config: AppConfig) -> Result<(), AppError> {
    let read_sql = config.colums.join(",");
    let read_sql = format!("SELECT {} FROM {} ", read_sql, config.sqlite_table);
    //读取sqlite库
    let insert_count: AtomicU32 = AtomicU32::new(0);

    // 创建运行时
    let r = create_runtime()?;
    //读取db文件
    //读取文件夹下的文件
    let files = r.block_on(dir::children_files(&config.sqlite_dir));
    info!("db文件数量:{}", files.len());
    //mysql客户端
    let client_r: Result<mysql::MysqlClient, mysql::MyslqError> =
        mysql::MysqlClient::new(&config.mysql_url);
    let client = check_result(client_r)?;

    info!("mysql连接成功");

    // 之前导入过的db文件
    let mut histroy = String::new();
    if file_util::exist(HISTORY_TXT) {
        histroy = check_result(file_util::read_utf8_file_str(HISTORY_TXT))?;
    }
    //读取数据添加到mysql
    for db_file in files {
        if !db_file.ends_with(".db") {
            continue;
        }
        let mut file_name = db_file.clone();
        // 获取文件名
        let filename_o: Option<String> = file_util::get_filename(&db_file);

        if let Some(fname) = filename_o {
            file_name = fname;
        }
        let file_name = file_name.as_str();

        if histroy.contains(file_name) {
            info!("跳过(已导入过):{}", file_name);
            continue;
        }
        info!("读取文件:{}", file_name);
        //查询当前db文件下的数据
        let mut read_s = rust_common::date::get_sys_timestamp_millis();
        // 按批次查询数据

        //保存添加结果
        // let mut insert_handles = Vec::new();
        let r = sqlite::query_data_by_batch(
            &db_file,
            &read_sql,
            &config.up_column,
            config.batch_size, //每次查多少条
            |list| {
                //查询到数据后的回调
                if list.len() == 0 {
                    return;
                }

                let len = list.len();
                let read_e = rust_common::date::get_sys_timestamp_millis();

                info!(
                    "读取数量:{}, 读取耗时:{}ms, 已添加数量:{}",
                    len,
                    (read_e - read_s),
                    insert_count.fetch_add(len as u32, Ordering::SeqCst)
                );

                info!("开始添加到mysq");

                let insert_r = r.block_on(client.save_batch_data_by_sql(
                    &config.mysql_table,
                    &config.colums,
                    list,
                ));

                let insert_ee = rust_common::date::get_sys_timestamp_millis();
                match insert_r {
                    Ok(()) => {
                        info!("添加成功,耗时:{}ms", (insert_ee - read_e))
                    }
                    Err(e) => {
                        error!("添加失败:{}", e)
                    }
                }

                read_s = rust_common::date::get_sys_timestamp_millis();
            },
        );

        let _ = check_result(r)?;

        let r = file_util::append_str(HISTORY_TXT, format!("{}\n", file_name).as_str());
        check_result(r)?;
        info!("写入历史记录:{}", file_name);
    }
    info!("关闭mysql连接");
    check_result(r.block_on(client.disconnect()))?;

    info!("总添加数量:{}", insert_count.fetch_add(0, Ordering::SeqCst));
    Ok(())
}

/// @Author: DengLibin
/// @Date: Create in 2023-11-01 17:07:53
/// @Description:
fn run() -> Result<(), AppError> {
    //日志初始化
    let r = log_util::init_log("log4rs.yaml");
    check_result(r)?;
    let r = config::read_yaml_config("app.yaml");

    match r {
        Ok(config) => {
            return sqlite2mysql(config);
        }
        Err(err) => {
            return Err(err);
        }
    }
}
fn main() {
    let s = rust_common::date::get_sys_timestamp_millis();
    let r = run();
    match r {
        Ok(()) => {
            info!("完成!")
        }
        Err(err) => {
            info!("异常:{}", err);
        }
    }
    let e = rust_common::date::get_sys_timestamp_millis();
    info!("耗时：{}ms", (e - s));
}
